/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.shortlink.mq.consumer;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.shortlink.common.exception.ServiceException;
import com.shortlink.mapper.*;
import com.shortlink.mq.idempotent.MessageQueueIdempotentHandler;
import com.shortlink.pojos.dto.dataobject.ShortLinkStatsRecordDO;
import com.shortlink.pojos.entity.*;
import com.shortlink.utils.BeanCopyUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

import static com.shortlink.common.constants.LinkConstant.AMAP_REMOTE_URL;

/**
 * 短链接监控状态保存消息队列消费者

 */
@Slf4j
@Component
@RequiredArgsConstructor
public class ShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {


    private final RedissonClient redissonClient;
    private final StringRedisTemplate stringRedisTemplate;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkMapper linkMapper;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @Value("${shortlink.locale.key}")
    private String localeKey;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String stream = message.getStream();
        RecordId id = message.getId();
        if(messageQueueIdempotentHandler.isMessageBeingConsumed(id.toString())){
            // 为应对主机挂了导致本该消费的消息没被消费的情况
            if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        try {
            // 处理逻辑
            Map<String, String> producerMap = message.getValue();
            String json=producerMap.get("statsRecord");
            ShortLinkStatsRecordDO statsRecord = JSON.parseObject(json, ShortLinkStatsRecordDO.class);
            actualSaveShortLinkStats(statsRecord);
            // 正常在删除消息前需要使用XACK命令
            // 使用 XACK 确认消息后，Redis 会将消息从待处理列表（Pending List）中移除，确保消息至少被处理一次。
            // 但由于幂等处理能够保证消息至少被消费一次，就没有用 XACK
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        }
        catch (Throwable throwable){// Throwable覆盖范围比Exception大
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            log.error("记录短链接监控消费异常", throwable);
            throw throwable;
        }
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }

    public void actualSaveShortLinkStats(ShortLinkStatsRecordDO requestParam){

        Date currentDate = requestParam.getCurrentDate();
        int hour = DateUtil.hour(currentDate, true);
        Week week = DateUtil.dayOfWeekEnum(currentDate);
        int weekValue = week.getValue()-1;
        Map<String, Object> localeParamMap = new HashMap<>();
        localeParamMap.put("key", localeKey);
        localeParamMap.put("ip", requestParam.getIp());
        String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);
        JSONObject localeResultObj = JSON.parseObject(localeResultStr);
        String infoCode = localeResultObj.getString("infocode");
        String actualProvince = "未知";
        if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")){
            String province = localeResultObj.getString("province");
            actualProvince=StrUtil.equals(province, "[]")? actualProvince:province;
        }

        LinkAccessLogsDO linkAccessLogsDO = BeanCopyUtils.copyBean(requestParam, LinkAccessLogsDO.class);
        linkAccessLogsDO.setProvince(actualProvince);
        linkAccessLogsMapper.insert(linkAccessLogsDO);

        LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                .shortUri(requestParam.getShortUri())
                .date(requestParam.getToday())
                .pv(1)
                .uv(requestParam.getUvFirstFlag()?1:0)
                .uip(requestParam.getUipFirstFlag()?1:0)
                .hour(hour)
                .weekday(weekValue)
                .build();
        linkAccessStatsMapper.shortLinkAccessStats(linkAccessStatsDO);

        LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                .shortUri(requestParam.getShortUri())
                .date(requestParam.getToday())
                .cnt(1)
                .browser(requestParam.getBrowser())
                .build();
        linkBrowserStatsMapper.shortLinkBrowserStats(linkBrowserStatsDO);

        LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                .shortUri(requestParam.getShortUri())
                .date(requestParam.getToday())
                .cnt(1)
                .device(requestParam.getDevice())
                .build();
        linkDeviceStatsMapper.shortLinkDeviceStats(linkDeviceStatsDO);

        LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                .shortUri(requestParam.getShortUri())
                .date(requestParam.getToday())
                .cnt(1)
                .province(actualProvince)
                .build();
        linkLocaleStatsMapper.shortLinkLocaleStats(linkLocaleStatsDO);

        LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                .shortUri(requestParam.getShortUri())
                .date(requestParam.getToday())
                .cnt(1)
                .network(requestParam.getNetwork())
                .build();
        linkNetworkStatsMapper.shortLinkNetworkStats(linkNetworkStatsDO);

        LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                .shortUri(requestParam.getShortUri())
                .date(requestParam.getToday())
                .cnt(1)
                .os(requestParam.getOs())
                .build();
        linkOsStatsMapper.shortLinkOsStats(linkOsStatsDO);

        linkMapper.incrementStats(requestParam.getShortUri(), 1, requestParam.getUvFirstFlag() ? 1 : 0, requestParam.getUipFirstFlag() ? 1 : 0);

    }
}
